package com.inspur.sc.springboot.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.inspur.sc.springboot.vo.HBaseColumnVO;
import com.inspur.sc.springboot.vo.HQueryVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class HbaseUtil {

    @Autowired
    private HbaseTemplate hbaseTemplate;

    /**
     * 写数据
     * @return
     */
    public Object execute(HQueryVO query) {
        if(StringUtils.isBlank(query.getRow()) || query.getColumns().isEmpty()){
            return null;
        }
        return hbaseTemplate.execute(query.getTable(), new TableCallback<Object>() {
            @SuppressWarnings("deprecation")
            @Override
            public Object doInTable(HTableInterface table) throws Throwable {
                try {
                    byte[] rowkey = query.getRow().getBytes();
                    Put put = new Put(rowkey);
                    for(HBaseColumnVO col:query.getColumns()){
                        put.addColumn(Bytes.toBytes(col.getFamily()), Bytes.toBytes(col.getQualifier()),
                                Bytes.toBytes(col.getValue()));
                    }
                    table.put(put);
                } catch (Exception e) {
                    log.warn("==> hbase get object fail> "+query.getRow());
                }
                return null;
            }

        });
    }


    /**
     * 通过表名和key获取一行数据
     * @param query
     * @param c
     * @param <T>
     * @return
     */
    public <T> T get(HQueryVO query, Class<T> c) {

        if(StringUtils.isBlank(query.getTable()) || StringUtils.isBlank(query.getRow())){
            return null;
        }

        return hbaseTemplate.get(query.getTable(), query.getRow(), new RowMapper<T>() {
            public T mapRow(Result result, int rowNum) throws Exception {
                List<Cell> ceList = result.listCells();
                T item=c.newInstance();
                JSONObject obj = new JSONObject();
                if (ceList != null && ceList.size() > 0) {
                    for (Cell cell : ceList) {
                        obj.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
                                cell.getQualifierLength()),
                                Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                    }
                }else{
                    return null;
                }
                item= JSON.parseObject(obj.toJSONString(), c);
                return item;
            }
        });
    }


    /**
     * 通过表名 key 和 列族 和列 获取一个数据
     *
     * @param query
     * @return
     */
    public String getColumn(HQueryVO query) {

        if(StringUtils.isBlank(query.getTable()) || StringUtils.isBlank(query.getRow())
                || StringUtils.isBlank(query.getFamily()) || StringUtils.isBlank(query.getQualifier())){
            return null;
        }

        return hbaseTemplate.get(query.getTable(), query.getRow(), query.getFamily(), query.getQualifier(), new RowMapper<String>() {
            public String mapRow(Result result, int rowNum) throws Exception {
                List<Cell> ceList = result.listCells();
                String res = "";
                if (ceList != null && ceList.size() > 0) {
                    for (Cell cell : ceList) {
                        res = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    }
                }
                return res;
            }
        });
    }

    /**
     * 通过表名，开始行键和结束行键获取数据
     *
     * @param query
     * @return
     */
    public <T> List<T> find(HQueryVO query,Class<T> c) {
        //如果未设置scan,设置scan
        if (query.getScan() == null) {

            if(query.getScan()==null){
                query.setScan(new Scan());
                query.getScan().setStartRow(Bytes.toBytes(query.getStartRow()));
                query.getScan().setStopRow(Bytes.toBytes(query.getStopRow()));
            }
        }

        //设置缓存
        query.getScan().setCacheBlocks(false);
        query.getScan().setCaching(2000);

        return hbaseTemplate.find(query.getTable(), query.getScan(), new RowMapper<T>() {
            @Override
            public T mapRow(Result result, int rowNum) throws Exception {

                List<Cell> ceList = result.listCells();
                JSONObject obj = new JSONObject();
                T item =c.newInstance();
                if (ceList != null && ceList.size() > 0) {
                    for (Cell cell : ceList) {

                        String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
                                cell.getValueLength());

                        String quali = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
                                cell.getQualifierLength());
                        if(value.startsWith("[")){
                            obj.put(quali, JSONArray.parseArray(value));
                        }else{
                            obj.put(quali, value);
                        }
                    }
                }
                item =JSON.parseObject(obj.toJSONString(), c);
                return item;
            }

        });
    }

    public void delete(HQueryVO query){
        hbaseTemplate.delete(query.getTable(), query.getRow(), query.getFamily());
    }
}
